package com.hex.ds.hdtp.core.app.data.transfer.service.impl;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.hex.ds.hdtp.core.app.common.entity.BasicTransferInfo;
import com.hex.ds.hdtp.core.app.common.page.PageData;
import com.hex.ds.hdtp.core.app.common.page.converter.IPageAndPageDtoConverter;
import com.hex.ds.hdtp.core.app.common.page.dto.PageDto;
import com.hex.ds.hdtp.core.app.data.table.dto.response.TableTransferFieldMapResponseDto;
import com.hex.ds.hdtp.core.app.data.table.dto.response.TableTransferListResponseDto;
import com.hex.ds.hdtp.core.app.data.table.service.ITableTransferFieldMapService;
import com.hex.ds.hdtp.core.app.data.table.service.ITableTransferListService;
import com.hex.ds.hdtp.core.app.data.transfer.converter.IDataTransferSqlStructureConverter;
import com.hex.ds.hdtp.core.app.data.transfer.converter.IDataTransferTaskConverter;
import com.hex.ds.hdtp.core.app.data.transfer.dto.DataTransferSqlStructureDto;
import com.hex.ds.hdtp.core.app.data.transfer.dto.request.DataTransferListRequestDto;
import com.hex.ds.hdtp.core.app.data.transfer.dto.request.DataTransferTaskRequestDto;
import com.hex.ds.hdtp.core.app.data.transfer.dto.response.DataTransferDownwardNumResponseDto;
import com.hex.ds.hdtp.core.app.data.transfer.dto.response.DataTransferListResponseDto;
import com.hex.ds.hdtp.core.app.data.transfer.dto.response.DataTransferTaskResponseDto;
import com.hex.ds.hdtp.core.app.data.transfer.entry.SourceInfo;
import com.hex.ds.hdtp.core.app.data.transfer.handler.impl.DataTransferSelectHandlerContext;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferListService;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferTaskService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceConfInfoService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceInfoService;
import com.hex.ds.hdtp.core.inf.common.page.Page;
import com.hex.ds.hdtp.core.inf.data.contrast.service.IDataInfoService;
import com.hex.ds.hdtp.core.inf.data.contrast.service.impl.DataInfoSelectHandlerContext;
import com.hex.ds.hdtp.core.inf.data.transfer.po.DataTransferTaskPo;
import com.hex.ds.hdtp.core.inf.data.transfer.repository.IDataTransferTaskRepository;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceInfoPo;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @Package: com.hex.ds.hdtp.core.app.data.transfer.service.impl
 * @ClassName DataTransferTaskService
 * @Description: 数据迁移任务-业务接口实现
 * @Author: wz.li
 * @Date 2023/8/17 20:00
 * @Version v2.0.2
 **/
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class DataTransferTaskService implements IDataTransferTaskService {

    private final IDataTransferTaskRepository dataTransferTaskRepository;

    private final IDataTransferTaskConverter converter;

    private final IPageAndPageDtoConverter pageAndPageDtoConverter;

    private final IDataTransferListService dataTransferListService;

    private final ISourceInfoService sourceInfoService;

    private final ITableTransferFieldMapService tableTransferFieldMapService;

    private final IDataTransferSqlStructureConverter dataTransferSqlStructureConverter;

    private final ITableTransferListService tableTransferListService;

    private final DataTransferSelectHandlerContext dataTransferSelectHandlerContext;

    private final ISourceConfInfoService sourceConfInfoService;

    private final DataInfoSelectHandlerContext dataInfoSelectHandlerContext;

    private IDataInfoService getIDataInfoService(String type) {
        return dataInfoSelectHandlerContext.getDataInfoService(type);
    }


    /*
     * @Method: queryListByPage <br>
     * @Param: [dataTransferTaskRequestDto, pageDto] <br>
     * @Return: com.hex.ds.hdtp.core.app.common.page.PageData <br>
     * @Description：分页查询<br>
     * @Author： wz.li<br>
     * @Date： 2023/8/17 20:03 <br>
     * @Version： V2.0.2<br>
     */
    @Override
    public PageData queryListByPage(DataTransferTaskRequestDto dataTransferTaskRequestDto, PageDto pageDto) {
        final Page<DataTransferTaskPo> pageResult = dataTransferTaskRepository.queryListByPage(
                converter.toPo(dataTransferTaskRequestDto), pageAndPageDtoConverter.toPage(pageDto));
        List<DataTransferTaskResponseDto> responseDtos = converter.toDtoList(pageResult.getData());
        return new PageData(responseDtos, pageResult.getTotal());
    }

    /*
     * @Method: add <br>
     * @Param: [dataTransferTaskRequestDto] <br>
     * @Return: void <br>
     * @Description：新增<br>
     * @Author： wz.li<br>
     * @Date： 2023/8/17 20:06 <br>
     * @Version： V2.0.2<br>
     */
    @Override
    public void add(DataTransferTaskRequestDto dataTransferTaskRequestDto) {
        addDataTransferTask(dataTransferTaskRequestDto);
    }

    /**
     * @Method addDataTransferTask <br>
     * @Param
     * @Param dataTransferTaskRequestDto <br>
     * @Return java.lang.String <br>
     * @Description 添加数据迁移任务<br>
     * @Author jian.qiao<br>
     * @Date 2023/10/27 18:22 <br>
     * @Version V1.0<br>
     */
    @Override
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public String addDataTransferTask(DataTransferTaskRequestDto dataTransferTaskRequestDto) {
        DataTransferTaskPo dataTransferTaskPo = converter.toPo(dataTransferTaskRequestDto);
        addTask(dataTransferTaskPo);
        try {
            addDataTransferList(dataTransferTaskPo.getPkId(), dataTransferTaskRequestDto.getTableList());
            toDataTransfer(dataTransferTaskRequestDto.getBizDateList(), dataTransferTaskPo.getPkId(), dataTransferTaskRequestDto.isDataAnalysis(), dataTransferTaskRequestDto.getDataTransferType());
        } catch (Exception e) {
            log.error("新增数据迁移清单并启动失败: {}", e.getMessage(), e);
            throw new RuntimeException("新增数据迁移清单并启动失败" + e.getMessage());
        }
        return dataTransferTaskPo.getPkId();
    }


    /*
     * @Method: toDataTransfer <br>
     * @Param: [bizDates, dataTransferTaskPkId, isDataAnalysis] <br>
     * @Return: void <br>
     * @Description：执行数据迁移任务<br>
     * @Author： wz.li<br>
     * @Date： 2023/8/24 17:52 <br>
     * @Version： V2.0.2<br>
     */
    private void toDataTransfer(List<DateTime> bizDates, String dataTransferTaskPkId, boolean isDataAnalysis, String transferType) {
        List<DataTransferListResponseDto> dataTransferList = dataTransferListService.queryListByTaskId(dataTransferTaskPkId);
        for (DataTransferListResponseDto dataTransferListResponseDto : dataTransferList) {
            DataTransferSqlStructureDto dataTransferSqlStructureDto = buildDataTransferSqlStructureDto(dataTransferListResponseDto);
            dataTransferSqlStructureDto.setBizDates(bizDates);
            if (transferType.equals("File")) {
                dataTransferSelectHandlerContext.getDataInfoFileService(dataTransferSqlStructureDto.getSourceDatabaseType() + "_" + dataTransferSqlStructureDto.getTargetDatabaseType() + "_FILE").dataFileTransfer(dataTransferSqlStructureDto, isDataAnalysis);
            } else if (transferType.equals("Jdbc")) {
                dataTransferSelectHandlerContext.getDataInfoJdbcService(dataTransferSqlStructureDto.getSourceDatabaseType() + "_" + dataTransferSqlStructureDto.getTargetDatabaseType() + "_JDBC").dataJdbcTransfer(dataTransferSqlStructureDto, isDataAnalysis);
            } else {
                dataTransferSelectHandlerContext.getDataInfoFlinkService(dataTransferSqlStructureDto.getSourceDatabaseType() + "_" + dataTransferSqlStructureDto.getTargetDatabaseType() + "_FLINK").dataFlinkTransfer(dataTransferSqlStructureDto, isDataAnalysis);
            }
        }
    }

    /*
     * @Method: buildDataTransferSqlStructureDto <br>
     * @Param: [dataTransferList] <br>
     * @Return: com.hex.ds.hdtp.core.app.data.transfer.dto.DataTransferSqlStructureDto <br>
     * @Description：构建数据迁移需要的参数<br>
     * @Author： wz.li<br>
     * @Date： 2023/8/24 17:52 <br>
     * @Version： V2.0.2<br>
     */
    private DataTransferSqlStructureDto buildDataTransferSqlStructureDto(DataTransferListResponseDto dataTransferList) {
        DataTransferSqlStructureDto dataTransferSqlStructureDto;
        SourceInfoPo sourceInfoPo = sourceInfoService.queryByPkId(dataTransferList.getSourceConnPkId());
        SourceInfoPo targetInfoPo = sourceInfoService.queryByPkId(dataTransferList.getTargetConnPkId());
        if (dataTransferList.getTableTransferListPkId() != null) {
            // 依赖表迁移构参数
            TableTransferListResponseDto tableTransferList = tableTransferListService.queryById(dataTransferList.getTableTransferListPkId());
            List<TableTransferFieldMapResponseDto> tableTransferFieldMap = tableTransferFieldMapService.queryListByListPkId(dataTransferList.getTableTransferListPkId());
            dataTransferSqlStructureDto = dataTransferSqlStructureConverter.toDto(dataTransferList, sourceInfoPo, targetInfoPo, tableTransferFieldMap);
            dataTransferSqlStructureDto.setPartitionFieldNameList(tableTransferList.getPartitionFieldNameList());
        } else {
            // 不依赖表迁移构参数
            BasicTransferInfo transferInfo = dataTransferSqlStructureConverter.toCrossDbContrastCommonDto(dataTransferList, sourceInfoPo, targetInfoPo);
            List<TableTransferFieldMapResponseDto> tableTransferFieldMaps = sourceInfoService.getFieldMaps(transferInfo);
            dataTransferSqlStructureDto = dataTransferSqlStructureConverter.toDto(dataTransferList, sourceInfoPo, targetInfoPo, tableTransferFieldMaps);
            try {
                dataTransferSqlStructureDto.setPartitionFieldNameList(getIDataInfoService(dataTransferSqlStructureDto.getSourceDatabaseType())
                        .selectPartitionField(transferInfo.getSourceCode(), transferInfo.getSourceSchema(), transferInfo.getSourceTable()));
            } catch (Exception e) {
                log.info("获取分区字段失败");
            }
        }
        return dataTransferSqlStructureDto;
    }

    /*
     * @Method: addDataTransferList <br>
     * @Param: [dataTransferTaskPkId, tableList] <br>
     * @Return: void <br>
     * @Description：新增数据同步清单<br>
     * @Author： wz.li<br>
     * @Date： 2023/8/18 15:39 <br>
     * @Version： V2.0.2<br>
     */
    private void addDataTransferList(String dataTransferTaskPkId, List<DataTransferListRequestDto> tableList) {
        for (DataTransferListRequestDto dataTransferListRequestDto : tableList) {
            dataTransferListRequestDto.setDataTransferTaskPkId(dataTransferTaskPkId);
            dataTransferListService.add(dataTransferListRequestDto);
        }
    }

    /*
     * @Method: modifyById <br>
     * @Param: [dataTransferTaskRequestDto] <br>
     * @Return: void <br>
     * @Description：根据主键更新<br>
     * @Author： wz.li<br>
     * @Date： 2023/8/17 20:06 <br>
     * @Version： V2.0.2<br>
     */
    @Override
    public void modifyById(DataTransferTaskRequestDto dataTransferTaskRequestDto) {
        DataTransferTaskPo dataTransferTaskPo = converter.toPo(dataTransferTaskRequestDto);
        if (dataTransferTaskRequestDto.getPkId() != null) {
            if (dataTransferTaskRepository.modifyById(converter.toPo(dataTransferTaskRequestDto)) != 1) {
                throw new RuntimeException("更新数据迁移任务失败");
            }
        } else {
            addTask(dataTransferTaskPo);
        }
        try {
            updateDataTransferList(dataTransferTaskPo.getPkId(), dataTransferTaskRequestDto.getTableList());
        } catch (Exception e) {
            log.error("更新迁移清单失败: {}", e.getMessage());
            throw new RuntimeException("更新数据迁移清单失败" + e.getMessage());
        }
    }

    private void updateDataTransferList(String dataTransferTaskPkId, List<DataTransferListRequestDto> tableList) {
        // 删除旧数据
        dataTransferListService.queryListByTaskId(dataTransferTaskPkId)
                .forEach(dataTransferListResponseDto -> dataTransferListService.removeById(dataTransferListResponseDto.getPkId()));
        // 添加新数据
        tableList.stream()
                .peek(dataTransferListRequestDto -> dataTransferListRequestDto.setDataTransferTaskPkId(dataTransferTaskPkId))
                .forEach(dataTransferListService::add);
    }

    /*
     * @Method: removeById <br>
     * @Param: [pkId] <br>
     * @Return: void <br>
     * @Description：根据主键删除<br>
     * @Author： wz.li<br>
     * @Date： 2023/8/17 20:06 <br>
     * @Version： V2.0.2<br>
     */
    @Override
    public void removeById(String pkId) {
        if (dataTransferTaskRepository.removeById(pkId) != 1) {
            throw new RuntimeException("删除数据迁移任务失败");
        }
    }

    /*
     * @Method: modifySuccessNum <br>
     * @Param: [pkId] <br>
     * @Return: void <br>
     * @Description：更新数据同步任务成功数量<br>
     * @Author： wz.li<br>
     * @Date： 2023/8/22 19:36 <br>
     * @Version： V2.0.2<br>
     */
    @Override
    public void modifySuccessNum(String pkId) {
        dataTransferTaskRepository.modifySuccessNum(pkId);
    }

    /*
     * @Method: modifyFailNum <br>
     * @Param: [pkId] <br>
     * @Return: void <br>
     * @Description：更新数据同步任务失败数量<br>
     * @Author： wz.li<br>
     * @Date： 2023/8/22 19:36 <br>
     * @Version： V2.0.2<br>
     */
    @Override
    public void modifyFailNum(String pkId) {
        dataTransferTaskRepository.modifyFailNum(pkId);
    }

    /*
     * @Method: modifyStatus <br>
     * @Param: [pkId] <br>
     * @Return: void <br>
     * @Description：更新数据同步任务状态<br>
     * @Author： wz.li<br>
     * @Date： 2023/8/22 19:37 <br>
     * @Version： V2.0.2<br>
     */
    @Override
    public void modifyStatus(String pkId) {
        dataTransferTaskRepository.modifyStatus(DateUtil.now(), pkId);
    }

    /**
     * @Method queryDownNum
     * @Param
     * @Param taskIdList
     * @Return com.hex.ds.hdtp.core.app.data.transfer.dto.response.DataTransferDownwardNumRequestDto
     * @Description 下数 信息整合
     * @Author xin.zhao
     * @Date 2023/10/20 10:43
     * @Version V1.0
     */
    @Override
    public List<DataTransferDownwardNumResponseDto> queryDownNum(DataTransferTaskRequestDto dataTransferTaskRequestDto) {
        List<String> taskIdList = dataTransferTaskRepository.queryTaskId(converter.toPo(dataTransferTaskRequestDto));
        ArrayList<DataTransferDownwardNumResponseDto> result = new ArrayList<>();
        if (!taskIdList.isEmpty()) {
            //foreach 遍历查询
            for (String taskId : taskIdList) {

                //查询所有 列表信息
                List<DataTransferListResponseDto> dataTransferListResponseDtoList = dataTransferListService.queryListByTaskId(taskId);
                SourceInfo sourceInfo = null;
                try {
                    sourceInfo = getSourceInfo(dataTransferListResponseDtoList);
                } catch (IllegalAccessException e) {
                    throw new RuntimeException(e);
                } catch (InstantiationException e) {
                    throw new RuntimeException(e);
                }
                //整合信息
                DataTransferDownwardNumResponseDto dataTransferDownwardNumResponseDto = converter.toDataTransferDownwardNumRequestDto(taskId, dataTransferListResponseDtoList.get(0).getSourceConnPkId(), dataTransferListResponseDtoList, sourceInfo);
                result.add(dataTransferDownwardNumResponseDto);
            }
            return result;
        }
        return result;
    }


    @Override
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void batchRemoveById(List<String> pkIds) {
        for (String pkId : pkIds) {
            removeById(pkId);
        }
    }

    @Override
    public String addWithBackId(DataTransferTaskRequestDto dataTransferTaskRequestDto) {
        DataTransferTaskPo dataTransferTaskPo = converter.toPo(dataTransferTaskRequestDto);
        if (dataTransferTaskRequestDto.getPkId() != null) {
            if (dataTransferTaskRepository.modifyById(converter.toPo(dataTransferTaskRequestDto)) != 1) {
                throw new RuntimeException("更新数据迁移任务失败");
            }
        } else {
            addTask(dataTransferTaskPo);
        }
        try {
            updateDataTransferList(dataTransferTaskPo.getPkId(), dataTransferTaskRequestDto.getTableList());
            toDataTransfer(dataTransferTaskRequestDto.getBizDateList(), dataTransferTaskPo.getPkId(), dataTransferTaskRequestDto.isDataAnalysis(), dataTransferTaskRequestDto.getDataTransferType());
        } catch (Exception e) {
            log.error("更新数据迁移清单并启动失败: {}", e.getMessage(), e);
            throw new RuntimeException("更新数据迁移清单并启动失败" + e.getMessage());
        }
        return dataTransferTaskPo.getPkId();
    }

    /**
     * @Method addTask
     * @Param dataTransferTaskPo
     * @Return void
     * @Description 新增数据迁移任务
     * @Author gj.xu
     * @Date 2024/1/25 14:35
     * @Version V2.0
     */
    private void addTask(DataTransferTaskPo dataTransferTaskPo) {
        if (dataTransferTaskRepository.add(dataTransferTaskPo) != 1) {
            throw new RuntimeException("新增数据迁移任务失败");
        }
    }


    /**
     * @Method getSourceInfo
     * @Param
     * @Param dataTransferListResponseDtoList
     * @Return com.hex.ds.hdtp.core.app.data.transfer.entry.SourceInfo
     * @Description 获得sourceInfo信息
     * @Author xin.zhao
     * @Date 2023/10/20 18:13
     * @Version V1.0
     */
    private SourceInfo getSourceInfo(List<DataTransferListResponseDto> dataTransferListResponseDtoList) throws IllegalAccessException, InstantiationException {
        if (dataTransferListResponseDtoList.isEmpty()) {
            return new SourceInfo(); // 或者根据你的需求返回默认值
        }
        String sourceConnPkId = dataTransferListResponseDtoList.get(0).getSourceConnPkId();
        SourceInfoPo sourceInfoPo = sourceInfoService.queryByPkId(sourceConnPkId);
        List<SourceConfInfoPo> sourceConfInfoPos = sourceConfInfoService.queryListBySourceCode(sourceInfoPo.getSourceCode());
        Map<String, String> map = sourceConfInfoPos.stream()
                .filter(po -> Objects.nonNull(po.getConfName()) && Objects.nonNull(po.getConfValue()))
                .collect(Collectors.toMap(SourceConfInfoPo::getConfName, SourceConfInfoPo::getConfValue));

        SourceInfo sourceInfo = mapToObj(map, SourceInfo.class);
        sourceInfo.setSourceCode(sourceInfoPo.getSourceType());
        return sourceInfo;

    }

    /**
     * @Method mapToObj
     * @Param
     * @Param map
     * @Param classType
     * @Return T
     * @Description 反射 将map 转 object
     * @Author xin.zhao
     * @Date 2023/10/20 18:13
     * @Version V1.0
     */
    public static <T> T mapToObj(Map<String, String> map, Class<T> classType) throws IllegalAccessException, InstantiationException {
        T object = classType.newInstance();
        for (Map.Entry<String, String> entry : map.entrySet()) {
            String fieldName = entry.getKey();
            String value = entry.getValue();
            try {
                Field field = classType.getDeclaredField(fieldName);
                field.setAccessible(true);
                if (field.getType() == String.class) {
                    field.set(object, value);
                } else if (field.getType() == int.class) {
                    field.set(object, Integer.parseInt(value));
                }
            } catch (NoSuchFieldException e) {
                log.info("此属性转换不存在");
            }
        }
        return object;
    }

}
